package com.zhuzhe.mq.mqtype.rabbitmq.sender;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

import com.zhuzhe.mq.mqtype.rabbitmq.init.RabbitmqConfig;
import com.zhuzhe.mq.service.MQBaseService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

/**
 * @author whh
 * @date 2020/08/24 20:32
 */
public class RabbitmqSender extends MQBaseService
    implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    private static Logger log = LoggerFactory.getLogger("RabbitMQ");
    private RabbitTemplate rabbitTemplate;
    private List<String> allServicesList = new ArrayList<>();

    @Autowired
    private RabbitmqConfig rabbitmqConfig;

    @Autowired
    public RabbitmqSender(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        rabbitTemplate.setConfirmCallback(this); // rabbitTemplate如果为单例的话，那回调就是最后设置的内容
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            log.info("消息发送成功,确认处理完成:" + correlationData);
        } else {
            log.warn("消息发送成功,确认处理失败:" + correlationData);
        }

    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        // log.error("message:" + message + " 发送失败");
    }

    /**
     * 发送消息
     *
     * @param exchangeName
     * @param routingKey
     * @param message
     */
    private void sendMessage(String exchangeName, String routingKey, Object message) {
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        log.info("开始发送消息:" + correlationId + ":" + message + "");
        rabbitTemplate.convertAndSend(exchangeName, routingKey, JSON.toJSONString(message), correlationId);
    }

    @Override
    public void send(String service, String dataType, Object message, boolean sendAll) {
        if (!allServicesList.contains(service)) {// 第一次发送消息时 绑定交换机和队列
            rabbitmqConfig.bindQueueToExchange(service, false, true);
            allServicesList.add(service);
        }
        String exchangeName = "com.zhuzhe." + service + "";
        String routingKey = "com.zhuzhe." + service + "";
        if (sendAll) {
            exchangeName = "com.zhuzhe." + service + "All";
            routingKey = "";
        }
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("dataType", dataType);
        jsonObject.put("data", message);
        sendMessage(exchangeName, routingKey, jsonObject);

        // List<MQServiceProperties> nameList = super.getNameByDataType(dataType);
        // for (int i = 0; i < nameList.size(); i++) {
        // MQServiceProperties mqServiceProperties = nameList.get(i);
        // String exchangeName = mqServiceProperties.getExchange();
        // String routingKey = mqServiceProperties.getRoutingKey();
        // if (sendAll) {
        // exchangeName = mqServiceProperties.getExchangeAll();
        // routingKey = "";
        // }
        // JSONObject jsonObject = new JSONObject();
        // jsonObject.put("dataType", dataType);
        // jsonObject.put("data", message);
        // sendMessage(exchangeName, routingKey, jsonObject);
        // }
    }
}
